package com.chamber.business.mqtt.config;

import lombok.extern.slf4j.Slf4j;
import org.eclipse.paho.client.mqttv3.*;
import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;

@Slf4j
@Component
public class  MqttFactory {
    // 客户端
    public static MqttClient mqttClient = null;
    // 内存持久性模型
    private static MemoryPersistence memoryPersistence = null;
    // 连接参数
    private static MqttConnectOptions mqttConnectOptions = new MqttConnectOptions();
    @Resource
    private MqttConfigurationProperties mqttConfigurationProperties;
    @Resource
    private MqttCallback mqttCallback;

    public void init() {
        // 初始化连接参数
        mqttConnectOptions = new MqttConnectOptions();
        mqttConnectOptions.setServerURIs(mqttConfigurationProperties.getHostUrl().split(","));
        mqttConnectOptions.setUserName(mqttConfigurationProperties.getUsername());
        mqttConnectOptions.setPassword(mqttConfigurationProperties.getPassword().toCharArray());
        mqttConnectOptions.setCleanSession(mqttConfigurationProperties.isClearSession());
        mqttConnectOptions.setConnectionTimeout(mqttConfigurationProperties.getTimeout());
        mqttConnectOptions.setKeepAliveInterval(mqttConfigurationProperties.getKeepalive());
        // 持久化方式
        memoryPersistence = new MemoryPersistence();
        try {
            // 客户端
            mqttClient = new MqttClient(mqttConfigurationProperties.getHostUrl(), mqttConfigurationProperties.getSubClientId(), memoryPersistence);
            // 设置回调
            mqttClient.setCallback(mqttCallback);
            // 建立连接
            mqttClient.connect(mqttConnectOptions);
            if (mqttClient.isConnected()) {
                log.info("MQTT服务器连接成功！");
            }
            // 多主题订阅
            String[] arrTopic = mqttConfigurationProperties.getDefaultTopic().split(",");
            String[] qos_temp = mqttConfigurationProperties.getQos().split(",");
            int[] arrQos = new int[qos_temp.length];
            for (int i = 0; i < qos_temp.length; i++) {
                arrQos[i] = Integer.valueOf(qos_temp[i]);
            }
            // 订阅
            mqttClient.subscribe(arrTopic,arrQos);
        } catch (MqttException e) {
            throw new RuntimeException(e);
        }
    }

    /**
     * 关闭客户端
     */
    public void closeConnect()
    {
        // 关闭存储方式
        if (memoryPersistence != null) {
            try {
                memoryPersistence.close();
            } catch (MqttPersistenceException e) {
                log.error("mqttConnectOptions closing failure！");
            }
        };
        // 关闭连接
        if (mqttClient != null) {
            try {
                mqttClient.disconnect();
                mqttClient.close();
            } catch (MqttException e) {
                log.error("mqttClient closing failure！");
            }
        }
    }
    /**
     * MQTT客户端重连
     */
    public void reConnect() {
        // 开始重连
        if(mqttClient != null || !mqttClient.isConnected()) {
            if(mqttConnectOptions !=null ) {
                try {
                    mqttClient.connect(mqttConnectOptions);
                    if (mqttClient.isConnected()) {
                        log.info("MQTT服务器重连成功！");
                    }
                } catch (MqttException e) {
                    log.error("MQTT服务器重连异常：",e);
                }
            } else {
                log.error("mqttConnectOptions is null！");
            }
        } else {
            log.error("mqttClient is null or connect！");
        }
    }
    /**
     * MQTT发布消息
     *
     * @param pubTopic
     * @param message
     */
    public void publishMessage(String pubTopic,String message) {
        if(null != mqttClient && mqttClient.isConnected()) {
            MqttMessage mqttMessage = new MqttMessage();
            mqttMessage.setPayload(message.getBytes());
            MqttTopic topic = mqttClient.getTopic(pubTopic);
            if (null == topic) {
                log.error("topic not exist！");
            }
            try {
                MqttDeliveryToken publish = topic.publish(mqttMessage);
                if(!publish.isComplete()) {
                    log.debug("客户端id:"+mqttClient.getClientId()+"，消息发布成功：" + message.toString());
                }
            } catch (MqttException e) {
                log.error("MQTT客户端发布消息异常：",e);
            }
        } else {
            reConnect();
        }
    }

    /**
     * MQTT发布消息（加入服务质量）
     *
     * @param pubTopic
     * @param message
     * @param qos
     */
    public void publishMessage(String pubTopic,String message,int qos)
    {
        if(null != mqttClient && mqttClient.isConnected())
        {
            log.debug("客户端id:"+mqttClient.getClientId()+"，有新消息发布！");
            MqttMessage mqttMessage = new MqttMessage();
            mqttMessage.setQos(qos);
            mqttMessage.setPayload(message.getBytes());
            MqttTopic topic = mqttClient.getTopic(pubTopic);
            if (null == topic) {
                log.error("topic not exist！");
            }
            try {
                MqttDeliveryToken publish = topic.publish(mqttMessage);
                if(!publish.isComplete()) {
                    log.debug("客户端id:"+mqttClient.getClientId()+"，消息发布成功！");
                }
            } catch (MqttException e) {
                log.error("MQTT客户端发布消息异常：",e);
            }
        } else {
            reConnect();
        }
    }



}
